#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
import os

from logger_util import get_logger
from spark_2_4_3.spark_submit_executor import SparkSubmitExecutor

logger = get_logger()


def get_contends(path):
    with open(path) as file_object:
        contends = file_object.read()
    return contends

@logger.catch
def main():
    parser = argparse.ArgumentParser(prog='python run_yarn.py')
    parser.add_argument("--master", help='spark-submit master local | yarn', required=False)
    parser.add_argument('--deploy_mode', help='deploy_mode default cluster', required=False)
    parser.add_argument('--driver_memory', help='driver_memory default 1G', required=False)
    parser.add_argument('--driver_cores', help='driver_cores default 1', required=False)
    parser.add_argument('--executor_memory', help='executor_memory default 1G', required=False)
    parser.add_argument('--num_executors', help='num_executors default 2', required=False)
    parser.add_argument('--executor_cores', help='executor_cores default 2', required=False)
    parser.add_argument('--spark_home', help='spark_home default $SPARK_HOME', required=False)
    parser.add_argument('--yarn_conf_dir', help='yarn_conf_dir default $YARN_CONF_DIR', required=False)
    parser.add_argument('--main', help='main py file path', required=True)
    parser.add_argument('--py_files', help='py_files path, automatic compression, multiple splitting by ,',
                        required=False)
    args = vars(parser.parse_args())
    shell_file_path = SparkSubmitExecutor().env_conf(args['spark_home'], args['yarn_conf_dir']) \
        .yarn_conf(deploy_mode=args['deploy_mode'], driver_memory=args['driver_memory'],
                   driver_cores=args['driver_cores'], executor_memory=args['executor_memory'],
                   num_executors=args['num_executors'], executor_cores=args['executor_cores']) \
        .run_on_yarn(args['py_files'], args['main'])
    logger.info("============================================================================")
    logger.info("this shell file path is {0}".format(shell_file_path))
    logger.info("file content: \n" +
                "{0}".format(get_contends(shell_file_path)) +
                "```\n")
    logger.info("============================================================================")
    os.system("sh " + shell_file_path)


if __name__ == '__main__':
    main()
